package acc

import org.apache.spark.rdd.RDD
import org.apache.spark.util.{AccumulatorV2, DoubleAccumulator, LongAccumulator}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable





object RDD_ACC {
  //ACC累加器
  def main(args: Array[String]): Unit = {
    val sparkconf=new SparkConf().setMaster("local[*]").setAppName("persis")
    val sc=new SparkContext(sparkconf)
    val rdd: RDD[String] = sc.makeRDD(List("Hello", "Spark", "hellp"), 2)
    //reduce先分区内计算，再分区间计算，分区内和分区间的计算规则是相同的
//    val i: Int = rdd.reduce(_ + _)
//    println(i)

    //获取系统累加器，spark默认就提供了简单数据聚合的累加器
    val sumAcc: LongAccumulator = sc.longAccumulator("sum")
//    rdd.foreach(
//      num=>{
//        sumAcc.add(num)
//      }
//    )
//    println(sumAcc.value)
    //创建自定义的累加器
    val wcAacc=new MyAccumulator()
    //注册自定义的累加器
    sc.register(wcAacc,"wcAccumulator")
    rdd.foreach(
      word=>{
        //使用累加器进行数据累加
        wcAacc.add(word)
      }
    )
    println(wcAacc.value)

    sc.stop()

  }
  //自定义数据累加器,
  /**需创建累加器对象，向spark进行注册
   1、定义累加器需要继承AccumulatorV2
   2、定义累加器中的泛型（IN：累加器输入的数据类型，OUT:累加器返回的数据类型）
   3、
   * */

  class MyAccumulator extends AccumulatorV2[String,mutable.Map[String,Long]]{
    private var wcMap=mutable.Map[String,Long]()
    var map:mutable.Map[String,Long]=mutable.Map()
    override def isZero: Boolean = {
      wcMap.isEmpty
    }

    override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] ={
     new MyAccumulator()
    }

    override def reset(): Unit = {
      wcMap.clear()
    }

    override def add(word: String): Unit = {
      val newCnt=wcMap.getOrElse(word,0L)+1
      wcMap.update(word,newCnt)
    }

    override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = {
      val map1=map
      val map2=other.value
      //两个map合并
      map=map1.foldLeft(map2)(
        (innerMap,kv)=>{
          innerMap(kv._1)=innerMap.getOrElse(kv._1,0L)+kv._2
          innerMap
        }
      )

    }

    override def value: mutable.Map[String, Long] = {
      wcMap
    }
  }

}
